package com.atguigu.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.Constant;
import com.atguigu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

//3.下游需求同时需要左表与右表数据,但是为累加型需求,输出过的数据输出0;状态编程
public class Dws08_LeftJoin_3 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  //生产环境中,主题并行度与Kafka主题的分区数保持一致

        //TODO 2.读取Kafka DWD层 下单主题数据
        DataStreamSource<String> kafkaDS = env.fromSource(KafkaUtil.getKafkaSource(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "dws_order_user_count_230524"), WatermarkStrategy.noWatermarks(), "kafka-source");

        //TODO 3.过滤&转换为JSON对象
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    try {
                        JSONObject jsonObject = JSONObject.parseObject(value);
                        out.collect(jsonObject);
                    } catch (Exception e) {
                        System.out.println("脏数据：" + value);
                    }
                }
            }
        });

        //TODO 4.按照order_detail_id进行分组,去重数据;累加型需求,输出过的数据输出0
        SingleOutputStreamOperator<JSONObject> filterDS = jsonObjDS.keyBy(json -> json.getString("id"))
                .map(new RichMapFunction<JSONObject, JSONObject>() {

                    private ValueState<String> leftState;
                    private ValueState<String> rightState1;
                    private ValueState<String> rightState2;

                    @Override
                    public void open(Configuration parameters) throws Exception {

//                        StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.seconds(5))
//                                .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
//                                .build();
                        ValueStateDescriptor<String> leftStateDescriptor = new ValueStateDescriptor<>("left", String.class);
                        ValueStateDescriptor<String> rightStateDescriptor1 = new ValueStateDescriptor<>("right1", String.class);
                        ValueStateDescriptor<String> rightStateDescriptor2 = new ValueStateDescriptor<>("right2", String.class);

//                        leftStateDescriptor.enableTimeToLive(ttlConfig);
//                        rightStateDescriptor1.enableTimeToLive(ttlConfig);
//                        rightStateDescriptor2.enableTimeToLive(ttlConfig);

                        leftState = getRuntimeContext().getState(leftStateDescriptor);
                        rightState1 = getRuntimeContext().getState(rightStateDescriptor1);
                        rightState2 = getRuntimeContext().getState(rightStateDescriptor2);
                    }

                    //order_detail_id order_id activity_id coupon_id
                    @Override
                    public JSONObject map(JSONObject value) throws Exception {

                        String leftStateValue = leftState.value();
                        String rightStateValue1 = rightState1.value();
                        String rightStateValue2 = rightState2.value();

                        JSONObject jsonObject = new JSONObject();
                        jsonObject.put("order_detail_id", value.getString("id"));

                        if (leftStateValue == null) {
                            jsonObject.put("order_id", value.getString("order_id"));
                            leftState.update("1");
                        } else {
                            jsonObject.put("order_id", "0");
                        }

                        if (rightStateValue1 == null) {
                            String activityId = value.getString("activity_id");
                            if (activityId == null) {
                                jsonObject.put("activity_id", "0");
                            } else {
                                jsonObject.put("activity_id", activityId);
                                rightState1.update("1");
                            }
                        } else {
                            jsonObject.put("activity_id", "0");
                        }

                        if (rightStateValue2 == null) {
                            String activityId = value.getString("coupon_id");
                            if (activityId == null) {
                                jsonObject.put("coupon_id", "0");
                            } else {
                                jsonObject.put("coupon_id", activityId);
                                rightState2.update("1");
                            }
                        } else {
                            jsonObject.put("coupon_id", "0");
                        }

                        return jsonObject;
                    }
                });

        filterDS.print(">>>>>>>>>");

        env.execute();

    }
}
